22. 长三角零售销售预测（Walmart 方法对照）

from pathlib import Path  # 导入Path以选择跨平台数据根目录
import pandas as pd  # 导入Pandas用于财报时间序列处理
import numpy as np  # 导入NumPy用于有限值检查要求
from sklearn.linear_model import LinearRegression  # 导入可解释线性回归作为候选模型
from sklearn.metrics import r2_score  # 导入R方用于描述验证期拟合边界

linux_root = Path('/home/ubuntu/r2_data_mount/data')  # 指定Linux课程数据根目录
windows_root = Path(r'C:\qiufei\data')  # 指定Windows课程数据根目录
data_root = linux_root if linux_root.exists() else windows_root  # 选择可用数据根目录
financial_path = data_root / 'stock' / 'financial_statement.h5'  # 定位财务报表本地快照

is_real_data_available = financial_path.exists()  # 将数据状态与渲染成功明确分离
if not is_real_data_available:  # 数据缺失时立即报告暂停但保持课件可渲染
    print(f'未找到课程数据文件：{financial_path}；请从课程数据下载入口获取后重新运行。')  # 提示读者下载所需文件

if is_real_data_available:  # 只在真实本地快照存在时定义披露选择约定
    financials = pd.read_hdf(financial_path, key='financial_data', where="order_book_id == '600827.XSHG'", columns=['order_book_id','quarter','info_date','operating_revenue']).rename(columns={'order_book_id':'ts_code','quarter':'end_date','info_date':'ann_date','operating_revenue':'total_revenue'})  # 只读取目标公司的必需财报字段并标准化
    financials['ts_code'] = financials['ts_code'].str.replace('.XSHG','.SH',regex=False).str.replace('.XSHE','.SZ',regex=False)  # 统一证券代码后缀
    financials['end_date'] = pd.PeriodIndex(financials['end_date'].str.upper(), freq='Q').end_time.normalize()  # 将季度键转换为报告期末日期
    def select_latest_disclosure(frame,value_fields):  # 统一财报披露版本选择约定
        required={'ts_code','end_date',*value_fields}; versions=[c for c in ['ann_date','f_ann_date'] if c in frame.columns]  # 固定键与可接受版本字段
        if not required<=set(frame.columns): raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 缺列时停止
        work=frame.dropna(subset=list(required)).copy(); work['end_date']=pd.to_datetime(work['end_date'],errors='coerce')  # 清理必需字段
        for column in versions: work[column]=pd.to_datetime(work[column],errors='coerce')  # 解析披露日期
        keys=['ts_code','end_date']; work=work.dropna(subset=['end_date']); duplicate_mask=work.duplicated(keys,keep=False); before=int(duplicate_mask.sum())  # 统计选择前重复
        if before and (not versions or work.loc[duplicate_mask,versions].isna().all(axis=1).any()): raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 无版本依据时停止
        work['_version']=work[versions].max(axis=1) if versions else pd.NaT  # 合并双版本字段为排序键
        latest_version=work.groupby(keys)['_version'].transform('max') if versions else pd.Series(pd.NaT,index=work.index)  # 为每个公司报告期定位最晚披露时点
        latest_rows=work.loc[~duplicate_mask | work['_version'].eq(latest_version)].copy()  # 仅保留单例或并列最晚版本
        tied_mask=latest_rows.duplicated(keys,keep=False); tied_latest_rows=int(tied_mask.sum())  # 统计并列最晚版本行
        conflicting_latest_keys=int(latest_rows.loc[tied_mask].groupby(keys,dropna=False)[value_fields].nunique(dropna=False).gt(1).any(axis=1).sum()) if tied_latest_rows else 0  # 统计并列且数值冲突的披露键
        if conflicting_latest_keys: raise ValueError(f'输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 并列最新值冲突时请先检查数据后再继续
        selected=latest_rows.drop_duplicates(keys+value_fields).drop_duplicates(keys).drop(columns='_version',errors='ignore')  # 仅合并数值完全一致的并列行
        after=int(selected.duplicated(keys).sum()); audit={'duplicate_rows_before':before,'tied_latest_rows':tied_latest_rows,'conflicting_latest_keys':conflicting_latest_keys,'duplicate_rows_after':after,'version_fields':versions}  # 返回并列与冲突检查
        if after: raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 选择后仍重复则停止
        return selected,audit  # 返回唯一版本与检查依据

if is_real_data_available:  # 数据可用时执行披露选择与累计转单季
    bailian,version_audit = select_latest_disclosure(financials.loc[financials['ts_code'].eq('600827.SH')],['total_revenue'])  # 筛选百联并保留最晚披露版本
    print('披露数据版本检查:',version_audit)  # 报告选择前后重复与采用字段
    bailian = bailian.loc[bailian['end_date'].between('2015-01-01', '2025-12-31')].sort_values('end_date').copy()  # 固定课堂样本边界与时间顺序
    bailian['fiscal_year'] = bailian['end_date'].dt.year  # 构造财政年键
    bailian['quarter'] = bailian['end_date'].dt.quarter  # 构造财政季度键
    if bailian.duplicated(['fiscal_year','quarter']).any(): raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 公司期间键不唯一时停止
    prior_cumulative=bailian[['fiscal_year','quarter','total_revenue']].rename(columns={'total_revenue':'prior_cumulative_revenue'}).copy()  # 建立前一财政季度查找表
    prior_cumulative['quarter']=prior_cumulative['quarter']+1  # 将每期键前移到其紧邻后季
    bailian=bailian.merge(prior_cumulative,on=['fiscal_year','quarter'],how='left',validate='one_to_one')  # 按同年紧邻季度精确连接累计值
    missing_predecessor=bailian['quarter'].gt(1) & bailian['prior_cumulative_revenue'].isna()  # 识别Q2至Q4缺失紧邻前季的情形
    if missing_predecessor.any(): raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 累计转单季无法精确相减时停止
    bailian['quarter_revenue'] = np.where(bailian['quarter'].eq(1),bailian['total_revenue'],bailian['total_revenue']-bailian['prior_cumulative_revenue'])  # 用Q1累计值或紧邻累计差得到单季收入

if is_real_data_available:  # 数据可用时构造上年同季基线与时间特征
    prior_season=bailian[['fiscal_year','quarter','quarter_revenue']].rename(columns={'quarter_revenue':'seasonal_baseline'}).copy()  # 建立上年同季查找表
    prior_season['fiscal_year']=prior_season['fiscal_year']+1  # 将历史年键前移一年以匹配当期
    bailian=bailian.merge(prior_season,on=['fiscal_year','quarter'],how='left',validate='one_to_one')  # 按相同财政季度精确连接上年基线
    bailian['trend_index'] = range(len(bailian))  # 构造只使用时间顺序的趋势特征

if is_real_data_available:  # 仅对已成功读取的真实单季数据评估
    model_frame = bailian.dropna(subset=['quarter_revenue', 'seasonal_baseline']).copy()  # 排除缺少同季基线的记录
    model_features = pd.get_dummies(model_frame[['trend_index', 'quarter']], columns=['quarter'], drop_first=True)  # 结合趋势与季节虚拟变量
    training_mask = model_frame['end_date'].lt('2023-01-01')  # 禁止验证期信息进入训练
    validation_mask = model_frame['end_date'].between('2023-01-01', '2024-12-31')  # 固定同一验证窗口
    if training_mask.sum()<8 or validation_mask.sum()<2: raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 拟合前检查窗口
    if model_frame.loc[validation_mask,'quarter_revenue'].eq(0).any(): raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 零实际值禁止进入MAPE
    regression = LinearRegression().fit(model_features.loc[training_mask], model_frame.loc[training_mask, 'quarter_revenue'])  # 拟合可解释候选模型
    model_frame.loc[validation_mask, 'regression_forecast'] = regression.predict(model_features.loc[validation_mask])  # 生成验证期预测
    training_residuals = model_frame.loc[training_mask, 'quarter_revenue'] - regression.predict(model_features.loc[training_mask])  # 估计训练残差波动
    interval_half_width = 1.96 * training_residuals.std(ddof=1)  # 仅在独立同分布、近似正态同方差且忽略参数不确定性时作课堂95%近似
    validation = model_frame.loc[validation_mask].copy()  # 隔离用于决策的验证样本
    validation['prediction_lower'] = validation['regression_forecast'] - interval_half_width  # 计算近似区间下界
    validation['prediction_upper'] = validation['regression_forecast'] + interval_half_width  # 计算近似区间上界
    baseline_mape = ((validation['quarter_revenue'] - validation['seasonal_baseline']).abs() / validation['quarter_revenue'].abs()).mean()  # 计算季节朴素基线MAPE
    regression_mape = ((validation['quarter_revenue'] - validation['regression_forecast']).abs() / validation['quarter_revenue'].abs()).mean()  # 计算候选模型MAPE
    validation_r2 = r2_score(validation['quarter_revenue'], validation['regression_forecast'])  # 报告拟合边界而非决策收益
    if not np.isfinite([baseline_mape,regression_mape,validation_r2,interval_half_width]).all(): raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 非有限依据不得进入决策
    validation_display=validation[['end_date','quarter_revenue','seasonal_baseline','regression_forecast','prediction_lower','prediction_upper']].copy()  # 建立紧凑课堂展示表
    validation_display['end_date']=validation['end_date'].dt.year.astype(str)+'-Q'+validation['quarter'].astype(str)  # 用短季度标签替代长时间戳
    validation_display.columns=['季度','实际(亿元)','季节基线','趋势模型','区间下界','区间上界']  # 使用短列名避免输出换行
    validation_display.iloc[:,1:]=validation_display.iloc[:,1:].div(1e8).round(1)  # 统一为亿元并控制一位小数
    print(validation_display.to_string(index=False))  # 每个验证季度仅占一行
    print({'base_mape':round(baseline_mape,3),'model_mape':round(regression_mape,3),'R2':round(validation_r2,3),'95%半宽_亿元':round(interval_half_width/1e8,1)})  # 用简短键单行输出同一时间段指标